package com.tomdd.iotdbapplication.utils;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.StrUtil;
import com.tomdd.iotdbapplication.vo.TimeValueVO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.MessageFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * <p>
 * iotDB查询工具类
 * </p>
 *
 * @author tz
 * @since 2024-08-28 14:01
 */
@Component
@Slf4j
public class IotDBUtils {

    @Resource
    @Qualifier("sessionPool")
    SessionPool iotdbSession;

    @Value("${iotdb.time.interval}")
    private Integer interval;
    //
    @Value("${iotdb.time.unit}")
    private String unit;
    //
    @Value("${iotdb.time.pointCount}")
    private Integer pointCount;
    //
    @Value("${iotdb.nanosecond}")
    private boolean nanosecond;


    public long getTime() {
        if (nanosecond) {
            return 1000000;
        } else {
            return 1;
        }
    }

    private long getTimeStamp(RowRecord rowRecord) {
        return rowRecord.getTimestamp() / getTime();
    }


    public String[] getFiledAndTableName(String iotdbAddress) {
        int lastIndex = iotdbAddress.lastIndexOf(".");
        if (lastIndex < 0) {
            log.error("iotdb地址不合法,iotdbAddress:{}", iotdbAddress);
            throw new BusinessException("iotdb地址不合法");
        }
        String filed = iotdbAddress.substring(0, lastIndex);
        String tableName = iotdbAddress.substring(lastIndex + 1);
        return new String[]{filed, tableName};
    }


    /*
     * 写数据至iotdb
     */
    public void insertRecord(String deviceId, Long time, List<String> measurements, List<TSDataType> types, List<Object> values) {
        try {
            iotdbSession.insertRecord(deviceId, time * getTime(), measurements, types, values);
        } catch (Exception e) {
            log.error("写数据至IotDB失败", e);
        }
    }

    /**
     * 批量插入记录
     *
     * @param deviceIds    设备路径列表
     * @param timestamps   时间戳列表
     * @param measurements 测点列表
     * @param values       值列表
     */
    public void insertRecords(List<String> deviceIds, List<Long> timestamps, List<List<String>> measurements, List<List<TSDataType>> types, List<List<Object>> values) {
        try {
            List<Long> times = timestamps.stream().map(time -> time * getTime()).collect(Collectors.toList());
            iotdbSession.insertRecords(deviceIds, times, measurements, types, values);
            log.info("批量插入记录成功：deviceIds={}", deviceIds);
        } catch (Exception e) {
            log.error("写数据至IotDB失败", e);
        }

    }

    /**
     * 时间区间查询IotDb
     *
     * @param field     属性列
     * @param tableName 表名
     * @param startDate 开始时间
     * @param endDate   结束时间
     * @return 点位值集合
     */
    public List<TimeValueVO> queryIotDbByTimeInterval(String field, String tableName, Date startDate, Date endDate) {
        // 拼接sql 查询IotDb
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN);
        String startTime = simpleDateFormat.format(startDate);
        String endTime = simpleDateFormat.format(endDate);
        String sql = "SELECT FIRST_VALUE(" + field + ") FROM " + tableName + " GROUP BY ([ " + startTime + " , " + endTime + ")," + interval + unit + ")";
        return executeSql(sql);
    }

    /**
     * 时间查询IotDb
     *
     * @param field     属性列
     * @param tableName 表名
     * @param date      时间
     * @return 点位值集合
     */
    public List<TimeValueVO> queryIotDbByTime(String field, String tableName, Date date) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN);
        String time = simpleDateFormat.format(date);
        String sql = "SELECT " + field + " FROM " + tableName + " WHERE TIME >= " + time + " LIMIT 1";
        return executeSql(sql);
    }

    /**
     * 查询IotDb最新值
     *
     * @param field     属性列
     * @param tableName 表名
     * @return 点位值集合
     */
    public TimeValueVO queryIotDb(String field, String tableName) {
        String sql = "SELECT " + field + " FROM " + tableName + " ORDER BY TIME DESC LIMIT 1";
        List<TimeValueVO> list = executeSql(sql);
        if (CollUtil.isNotEmpty(list)) {
            return list.get(0);
        }
        return null;
    }

    /**
     * iotDB执行sql
     *
     * @param sql 需执行的sql
     * @return 点位值集合
     */
    public List<TimeValueVO> executeSql(String sql) {
        log.info("iotDB sql : {}", sql);
        List<TimeValueVO> timeValueVOList = new ArrayList<>();
        try (SessionDataSetWrapper sessionDataSet = iotdbSession.executeQueryStatement(sql)) {
            while (sessionDataSet.hasNext()) {
                RowRecord rowRecord = sessionDataSet.next();
                String stringValue = rowRecord.getFields().get(0).getStringValue();
                if ("null".equals(stringValue)) {
                    // null字符串解析为0
                    stringValue = "0";
                }
                TimeValueVO timeValueVO = new TimeValueVO(getTimeStamp(rowRecord), new BigDecimal(stringValue));
                timeValueVOList.add(timeValueVO);
            }
        } catch (Exception e) {
            log.error("queryIotDbByTimeInterval error", e);
        }
        return timeValueVOList;
    }

    /***
     * 根据path集合查询一段时间内的值（非聚合）
     * @param path
     * @param startTime
     * @param endTime
     * @return
     */
    public List<TimeValueVO> queryIntervalIotDB(String path, long startTime, Long endTime) {
        try {
            return queryIntervalIotDBAgg(path, startTime, endTime, "avg", 60, null
            );
        } catch (Exception e) {
            log.error("查询IotDB失败", e);
            throw new BusinessException("查询queryIntervalIotDB失败");
        }
    }


    /**
     * 根据path聚合查询出一个值（avg,max,min等）
     *
     * @param path
     * @param startTime
     * @param endTime
     */
    public BigDecimal queryAggregateValueIotDB(String path, long startTime, Long endTime, String aggregateMethod, BigDecimal th_low, BigDecimal th_high) {
        if (StrUtil.isBlank(path)) {
            return null;
        }
        SessionDataSetWrapper sessionDataSet = null;
        try {
//            startTime = 1000000 * startTime;
//            endTime = 1000000 * endTime;

            String measurement = path.substring(path.lastIndexOf(".") + 1);
            String realPath = path.substring(0, path.lastIndexOf("."));
            String sql = null;
            if (th_high == null) {
                if (th_low == null) {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s", aggregateMethod, measurement, realPath, startTime, endTime);
                } else {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s >= %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_low);
                }
            } else {
                if (th_low == null) {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s <= %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_high);
                } else {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s >= %s and %s <= %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_low, measurement, th_high);
                }
            }
            log.info(sql);

            sessionDataSet = iotdbSession.executeQueryStatement(sql);
            String stringValue = "";
            if (sessionDataSet.hasNext()) {
                stringValue = sessionDataSet.next().getFields().get(0).getStringValue();
            }

            if (StrUtil.isBlank(stringValue) || Objects.equals(stringValue, "null")) {
                return BigDecimal.ZERO;
            }

            sessionDataSet.close();
            return new BigDecimal(stringValue);

        } catch (Exception e) {
            log.error("queryAggregateValueIotDB根据path聚合查询出一个值失败", e);
            return null;

        } finally {
            if (sessionDataSet != null) {
                try {
                    sessionDataSet.close();
                } catch (Exception e) {
                    log.error("sessionDataSet资源为空，没有被正确关闭", e);
                    throw new BusinessException("sessionDataSet资源为空，没有被正确关闭");
                }
            }
        }
    }

    /**
     * 根据时间区间 和 采样数量 获取 aggSeconds-采样间隔-秒
     *
     * @param startOfPeriod 开始时间
     * @param endOfPeriod   结束时间
     * @return aggSeconds-采样间隔-秒
     */
    public int getAggSeconds(long startOfPeriod, long endOfPeriod) {
        long timeInterval = (endOfPeriod - startOfPeriod + 1) / 1000;
        return (int) timeInterval / pointCount;
    }


    /**
     * 计算在开始时间和结束时间段内，pointCount个点位需要间隔多少秒。
     *
     * @param startOfPeriod 开始时间戳（毫秒）
     * @param endOfPeriod   结束时间戳（毫秒）
     * @param pointCount    数据点的数量
     * @return 每个数据点之间的平均时间间隔（秒）
     */
    public int calculateIntervalInSeconds(long startOfPeriod, long endOfPeriod, int pointCount) {
        if (pointCount <= 0) {
            throw new IllegalArgumentException("Point count must be greater than 0");
        }

        long totalDurationInSeconds = (endOfPeriod - startOfPeriod) / 1000;
        // 如果只有一个点，则不需要间隔时间
        if (pointCount == 1) {
            return 0;
        }
        // 总时间间隔除以点数减一，得到每个点之间的平均间隔时间
        return (int) (totalDurationInSeconds / (pointCount - 1));
    }

    /***
     * 利用iotdb udf进行降采样查询一段时间内数据
     * @param path 完整的路径（path+measurement）
     * @param startTime 开始时间 毫秒时间戳
     * @param endTime 结束时间 毫秒时间戳
     * @param aggMethod 聚合方法 avg,max,min等
     * @param aggSeconds 采样间隔 秒
     */
    public List<TimeValueVO> queryIntervalIotDBAgg(String path, long startTime, long endTime, String aggMethod, int aggSeconds, Integer scale) {
        if (StrUtil.isBlank(path)) {
            return null;
        }
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            startTime = startTime * getTime();
            endTime = endTime * getTime();
            aggMethod = aggMethod == null ? "avg" : aggMethod;
            String measurement = path.substring(path.lastIndexOf(".") + 1);
            String realPath = path.substring(0, path.lastIndexOf("."));
            //计算udf参数proportion.假设iotdb存储的数据
            String sql = MessageFormat.format("select {0}({1}) from {2}  group by ([{3},{4}),{5}s)", aggMethod, measurement, realPath, String.valueOf(startTime), String.valueOf(endTime), String.valueOf(aggSeconds));
            log.info("queryIntervalIotDBAgg sql:{}", sql);
            sessionDataSetWrapper = iotdbSession.executeQueryStatement(sql);
            List<TimeValueVO> res = new ArrayList<>();
            while (sessionDataSetWrapper.hasNext()) {
                RowRecord rowRecord = sessionDataSetWrapper.next();
                long timeStamp = getTimeStamp(rowRecord);
                if (rowRecord.getFields().get(0).getDataType() == null || rowRecord.getFields().get(0).toString().equals("NaN")) {
                    res.add(new TimeValueVO(timeStamp, null));
                    continue;
                }
                String dataType = rowRecord.getFields().get(0).getDataType().toString();

                switch (dataType) {
                    case "BOOLEAN":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getBoolV()));
                        break;
                    case "INT32":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getIntV()));
                        break;
                    case "INT64":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getLongV()));
                        break;
                    case "FLOAT":
                        Object value = scale == null ? rowRecord.getFields().get(0).getFloatV() : new BigDecimal(String.valueOf(rowRecord.getFields().get(0).getFloatV())).setScale(scale, RoundingMode.HALF_UP);
                        res.add(new TimeValueVO(timeStamp, value));
                        break;
                    case "DOUBLE":
                        String stringValue = rowRecord.getFields().get(0).getStringValue();
                        if (StrUtil.isBlank(stringValue) || Objects.equals(stringValue, "null") || Objects.equals(stringValue, "NaN")) {
                            res.add(new TimeValueVO(timeStamp, null));
                        } else {
                            Object value2 = scale == null ? new BigDecimal(rowRecord.getFields().get(0).getStringValue()) : new BigDecimal(rowRecord.getFields().get(0).getStringValue()).setScale(scale, RoundingMode.HALF_UP);
                            res.add(new TimeValueVO(timeStamp, value2));
                        }
                        break;
                    case "TEXT":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getStringValue()));
                        break;
                    default:
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(1).getDoubleV()));
                        break;
                }
            }
//            sessionDataSetWrapper.close();
            return res;
        } catch (Exception e) {
            log.error(MessageFormat.format("queryIntervalIotDBAgg查询IotDB失败：path:{0},startTime:{1},endTime:{2},", path, startTime, endTime), e);
            return new ArrayList<TimeValueVO>();
//            throw new IotDBException(ErrorCodeUtil.buildCodeMsgDto("-1", MessageFormat.format("iotdb queryIntervalIotDBAgg查询失败,参数{0} {1} {2} {3} {45}", path, startTime, endTime, aggMethod, aggSeconds)));
        } finally {
            if (sessionDataSetWrapper != null) {
                sessionDataSetWrapper.close();
            }
        }
    }


    /***
     * 利用iotdb udf进行降采样查询一段时间内数据
     * @param path 完整的路径（path+measurement）
     * @param startTime 开始时间 毫秒时间戳
     * @param endTime 结束时间 毫秒时间戳
     * @param aggMethod 聚合方法 avg,max,min等
     * @param aggSeconds 采样间隔 秒
     */
    public BigDecimal queryIntervalIotDBAggSum(String path, long startTime, long endTime, String aggMethod, int aggSeconds, Integer scale) {
        if (StrUtil.isBlank(path)) {
            return null;
        }
        SessionDataSetWrapper sessionDataSetWrapper = null;
        BigDecimal zero = BigDecimal.ZERO;
        try {
            startTime = startTime * getTime();
            endTime = endTime * getTime();
            aggMethod = aggMethod == null ? "avg" : aggMethod;
            String measurement = path.substring(path.lastIndexOf(".") + 1);
            String realPath = path.substring(0, path.lastIndexOf("."));
            //计算udf参数proportion.假设iotdb存储的数据
            String sql = MessageFormat.format("select {0}({1}) from {2}  group by ([{3},{4}),{5}s)", aggMethod, measurement, realPath, String.valueOf(startTime), String.valueOf(endTime), String.valueOf(aggSeconds));
            log.info("queryIntervalIotDBAgg sql:{}", sql);
            sessionDataSetWrapper = iotdbSession.executeQueryStatement(sql);
            List<TimeValueVO> res = new ArrayList<>();
            while (sessionDataSetWrapper.hasNext()) {
                RowRecord rowRecord = sessionDataSetWrapper.next();
                long timeStamp = getTimeStamp(rowRecord);
                if (rowRecord.getFields().get(0).getDataType() == null || rowRecord.getFields().get(0).toString().equals("NaN")) {
                    res.add(new TimeValueVO(timeStamp, null));
                    continue;
                }
                String dataType = rowRecord.getFields().get(0).getDataType().toString();

                switch (dataType) {
                    case "BOOLEAN":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getBoolV()));
                        break;
                    case "INT32":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getIntV()));
                        break;
                    case "INT64":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getLongV()));
                        break;
                    case "FLOAT":
                        Object value = scale == null ? rowRecord.getFields().get(0).getFloatV() : new BigDecimal(String.valueOf(rowRecord.getFields().get(0).getFloatV())).setScale(scale, RoundingMode.HALF_UP);
                        res.add(new TimeValueVO(timeStamp, value));
                        break;
                    case "DOUBLE":
                        String stringValue = rowRecord.getFields().get(0).getStringValue();
                        if (StrUtil.isBlank(stringValue) || Objects.equals(stringValue, "null") || Objects.equals(stringValue, "NaN")) {
                            res.add(new TimeValueVO(timeStamp, null));
                        } else {
                            Object value2 = scale == null ? new BigDecimal(rowRecord.getFields().get(0).getStringValue()) : new BigDecimal(rowRecord.getFields().get(0).getStringValue()).setScale(scale, RoundingMode.HALF_UP);
                            res.add(new TimeValueVO(timeStamp, value2));
                        }
                        break;
                    case "TEXT":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getStringValue()));
                        break;
                    default:
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(1).getDoubleV()));
                        break;
                }
                zero = zero.add(BigDecimal.valueOf(rowRecord.getFields().get(1).getDoubleV()));
            }
//            sessionDataSetWrapper.close();
            return zero;
        } catch (Exception e) {
            log.error(MessageFormat.format("queryIntervalIotDBAgg查询IotDB失败：path:{0},startTime:{1},endTime:{2},", path, startTime, endTime), e);
            return zero;
//            throw new IotDBException(ErrorCodeUtil.buildCodeMsgDto("-1", MessageFormat.format("iotdb queryIntervalIotDBAgg查询失败,参数{0} {1} {2} {3} {45}", path, startTime, endTime, aggMethod, aggSeconds)));
        } finally {
            if (sessionDataSetWrapper != null) {
                sessionDataSetWrapper.close();
            }
        }
    }


    /**
     * 查询 IoTDB 数据库，返回指定时间间隔的结果列表。
     *
     * @param path      时间序列路径
     * @param startTime 开始时间戳
     * @param endTime   结束时间戳
     * @param interval  时间间隔（毫秒）
     * @return 时间值对象列表
     */
    public List<TimeValueVO> queryIntervalDataFromIotDB(String path, long startTime, long endTime, long interval, int decimalPlaces) {
        SessionDataSetWrapper sessionDataSetWrapper = null;

        String measurement = path.substring(path.lastIndexOf(".") + 1);
        String devicePath = path.substring(0, path.lastIndexOf("."));
        try {
            startTime = startTime * getTime();
            endTime = endTime * getTime();
            String sql = String.format("SELECT first_value(ROUND(%s, %d)) FROM %s GROUP BY ([%d, %d), %ds)",
                    measurement, decimalPlaces, devicePath, startTime, endTime, interval);
//            String sql = String.format("SELECT ROUND(%s, %d) FROM %s WHERE time > %d AND time < %d AND ((time - %d) %% %d) = 0",
//                    measurement, decimalPlaces, devicePath, startTime, endTime, startTime, interval);
            log.info("queryIntervalDataFromIotDB sql: {}", sql);
            sessionDataSetWrapper = iotdbSession.executeQueryStatement(sql);
            List<TimeValueVO> res = new ArrayList<>();
            while (sessionDataSetWrapper.hasNext()) {
                RowRecord rowRecord = sessionDataSetWrapper.next();
                long timeStamp = getTimeStamp(rowRecord);
                Field field = rowRecord.getFields().get(0);
                Object value = field.getObjectValue(field.getDataType());
                res.add(new TimeValueVO(timeStamp, value));
            }
            return res;
        } catch (Exception e) {
            log.error(String.format("queryIntervalDataFromIotDB查询IotDB失败：path:%s,startTime:%d,endTime:%d,interval:%d", path, startTime, endTime, interval), e);
            return new ArrayList<>();
        } finally {
            if (sessionDataSetWrapper != null) {
                sessionDataSetWrapper.close();
            }
        }
    }

    /**
     * 查询与时间最靠近值
     *
     * @param path
     * @param queryTime
     * @param scale
     * @return
     */
    public TimeValueVO queryNearestValueByTime(String path, long queryTime, Integer scale) {
        if (StrUtil.isBlank(path)) {
            return null;
        }
        SessionDataSetWrapper sessionDataSet = null;
        try {
            queryTime = queryTime * getTime();
            String measurement = path.substring(path.lastIndexOf(".") + 1);
            String realPath = path.substring(0, path.lastIndexOf("."));
            String sql = MessageFormat.format("select last_value( {0} ) from {1} where time <= {2}  ", measurement, realPath, String.valueOf(queryTime));
            log.info("queryNearestValueByTime sql:{}", sql);
            sessionDataSet = iotdbSession.executeQueryStatement(sql);
            List<TimeValueVO> res = new ArrayList<>();
            while (sessionDataSet.hasNext()) {
                RowRecord rowRecord = sessionDataSet.next();
                long timeStamp = 1000L * ((int) (getTimeStamp(rowRecord) / 1000000000));
                if (rowRecord.getFields().get(0).getDataType() == null) {
                    sessionDataSet.close();
                    return new TimeValueVO(queryTime, null);
                }
                String dataType = rowRecord.getFields().get(0).getDataType().toString();
                switch (dataType) {
                    case "BOOLEAN":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getBoolV()));
                        break;
                    case "INT32":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getIntV()));
                        break;
                    case "INT64":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getLongV()));
                        break;
                    case "FLOAT":
                        Object value = scale == null ? rowRecord.getFields().get(0).getFloatV() : new BigDecimal(String.valueOf(rowRecord.getFields().get(0).getFloatV())).setScale(scale, RoundingMode.HALF_UP);
                        res.add(new TimeValueVO(timeStamp, value));
                        break;
                    case "DOUBLE":
                        Object value2 = scale == null ? new BigDecimal(rowRecord.getFields().get(0).getStringValue()) : new BigDecimal(rowRecord.getFields().get(0).getStringValue()).setScale(scale, RoundingMode.HALF_UP);
                        res.add(new TimeValueVO(timeStamp, value2));
                        break;
                    case "TEXT":
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(0).getStringValue()));
                        break;
                    default:
                        res.add(new TimeValueVO(timeStamp, rowRecord.getFields().get(1).getDoubleV()));
                        break;
                }
            }
//            sessionDataSet.close();
            if (res.size() == 0) {
                return new TimeValueVO(queryTime, 0);
            }
            return res.get(0);

        } catch (Exception e) {
            log.error("查询IotDB失败：" + path, e);
            return new TimeValueVO(queryTime, 0);
        } finally {
            if (sessionDataSet != null) {
                try {
                    sessionDataSet.close();
                } catch (Exception e) {
                    log.error("sessionDataSet资源为空，没有被正确关闭", e);
                    throw new BusinessException("sessionDataSet资源为空，没有被正确关闭");
                }
            }
        }
    }

    /***
     * 根据storagepath删除数据timeseries
     * @param path
     */
    public void deleteStoragePath(String path) {
        try {
            String sql = String.format("delete timeseries %s", path);
            iotdbSession.executeNonQueryStatement(sql);
        } catch (Exception e) {
            log.error("删除IotDB失败", e);
        }
    }

    /**
     * 查询IotDb最新值,小于多少的
     *
     * @param field     属性列
     * @param tableName 表名
     * @return 点位值集合
     */
    public TimeValueVO queryIotDbLess(String field, String tableName, BigDecimal th_low) {
        String sql = "SELECT " + field + " FROM " + tableName + " WHERE " + field + " < " + th_low + "  ORDER BY TIME DESC LIMIT 1";
        List<TimeValueVO> list = executeSql(sql);
        if (CollUtil.isNotEmpty(list)) {
            return list.get(0);
        }
        return null;
    }

    /**
     * 查询IotDb最新值,小于多少的
     *
     * @return 点位值集合
     */
    public TimeValueVO queryIotDbLessByTime(String path, BigDecimal th_low, long endTime) {
        endTime = endTime * getTime();
        String measurement = path.substring(path.lastIndexOf(".") + 1);
        String devicePath = path.substring(0, path.lastIndexOf("."));
        String sql = MessageFormat.format("SELECT {0} FROM {1} WHERE {2} < {3} AND time <= {4} ORDER BY TIME DESC LIMIT 1", measurement, devicePath, measurement, th_low, String.valueOf(endTime));
        List<TimeValueVO> list = executeSql(sql);
        if (CollUtil.isNotEmpty(list)) {
            return list.get(0);
        }
        return null;
    }

    /**
     * 查询某段时间内的聚合值
     *
     * @param path              点位路径
     * @param aggregationMethod 聚合方法
     * @param startTime         开始时间
     * @param endTime           结束时间
     * @param decimalPlaces     保留小数个数
     * @return
     */
    public TimeValueVO executeAggregationQuery(String path, String aggregationMethod, long startTime, long endTime, int decimalPlaces) {
        startTime = startTime * getTime();
        endTime = endTime * getTime();
        String measurement = path.substring(path.lastIndexOf(".") + 1);
        String devicePath = path.substring(0, path.lastIndexOf("."));
        String sql = String.format("SELECT %s(%s) FROM %s WHERE time >= %d AND time <= %d AND %s = %s",
                aggregationMethod, measurement, devicePath, startTime, endTime, measurement, measurement);
        log.info("queryIntervalDataFromIotDB sql: {}", sql);
        List<TimeValueVO> list = executeSql(sql);

        if (list != null && !list.isEmpty()) {
            TimeValueVO result = list.get(0);
            Object value = result.getValue();
            if (value instanceof BigDecimal) {
                // Round the BigDecimal value to the specified number of decimal places
                BigDecimal roundedValue = ((BigDecimal) value).setScale(decimalPlaces, RoundingMode.HALF_UP);
                result.setValue(roundedValue);
            } else {
                // Handle the case where the value is not a BigDecimal
                // You may need to throw an exception or convert it to a BigDecimal
                throw new IllegalArgumentException("Value is not a BigDecimal");
            }
            return result;
        }
        return null;

    }

    /**
     * 查询一段时间内的值
     *
     * @param path
     * @param startTime
     * @param endTime
     * @return
     */
    public List<TimeValueVO> queryTimesAll(String path, long startTime, long endTime) {
        startTime = startTime * getTime();
        endTime = endTime * getTime();
        SessionDataSetWrapper sessionDataSetWrapper = null;

        String measurement = path.substring(path.lastIndexOf(".") + 1);
        String devicePath = path.substring(0, path.lastIndexOf("."));
        try {
            String sql = String.format("SELECT %s FROM %s WHERE time >= %d AND time <= %d",
                    measurement, devicePath, startTime, endTime);
            log.info("queryTimesAll sql: {}", sql);
            sessionDataSetWrapper = iotdbSession.executeQueryStatement(sql);
            List<TimeValueVO> res = new ArrayList<>();
            while (sessionDataSetWrapper.hasNext()) {
                RowRecord rowRecord = sessionDataSetWrapper.next();
                long timeStamp = getTimeStamp(rowRecord);
                Field field = rowRecord.getFields().get(0);
                Object value = field.getObjectValue(field.getDataType());
                res.add(new TimeValueVO(timeStamp, value));
            }
            return res;
        } catch (Exception e) {
            log.error(String.format("queryTimesAll查询IotDB失败：path:%s,startTime:%d,endTime:%d", path, startTime, endTime), e);
            return new ArrayList<>();
        } finally {
            if (sessionDataSetWrapper != null) {
                sessionDataSetWrapper.close();
            }
        }
    }

    /**
     * 查询一段时间内的最大值或最小值
     *
     * @param path                路径
     * @param aggregationFunction 聚合函数，例如MAX_VALUE  MIN_VALUE
     * @param startTime           开始时间
     * @param endTime             结束时间
     * @return
     */
    public BigDecimal queryMaxMinValue(String path, String aggregationFunction, long startTime, long endTime) {
        startTime = startTime * getTime();
        endTime = endTime * getTime();
        String measurement = path.substring(path.lastIndexOf(".") + 1);
        String devicePath = path.substring(0, path.lastIndexOf("."));
        String sql = String.format("SELECT %s(%s) FROM %s WHERE time > %d AND time < %d",
                aggregationFunction, measurement, devicePath, startTime, endTime);
        log.info("executeAggregationQuery sql: {}", sql);
        List<TimeValueVO> list = executeSql(sql); // Assume this method executes the SQL and returns a list of TimeValueVO

        if (list != null && !list.isEmpty()) {
            TimeValueVO result = list.get(0);
            Object value = result.getValue();
            if (value instanceof BigDecimal) {
                // Round the BigDecimal value to the specified number of decimal places
                BigDecimal roundedValue = ((BigDecimal) value);
                return roundedValue;
            } else {
                // Handle the case where the value is not a BigDecimal
                throw new IllegalArgumentException("Value is not a BigDecimal");
            }
        }
        return null;
    }

    /**
     * 根据最大值或最小值查询对应得时间戳
     *
     * @param path                路径
     * @param aggregationFunction 最大时间或最小时间得聚合函数
     * @param specificValue       最大值或最小值
     * @param startTime           开始时间
     * @param endTime             结束时间
     * @return
     */

    public Long queryMaxMinTime(String path, String aggregationFunction, BigDecimal specificValue, long startTime, long endTime) {
        startTime = startTime * getTime();
        endTime = endTime * getTime();
        String measurement = path.substring(path.lastIndexOf(".") + 1);
        String devicePath = path.substring(0, path.lastIndexOf("."));
        String sql = String.format("SELECT %s(%s) FROM %s WHERE %s = %s AND time >= %d AND time <= %d",
                aggregationFunction, measurement, devicePath, measurement, specificValue, startTime, endTime);
        log.info("executeCustomAggregationQuery sql: {}", sql);
        List<TimeValueVO> list = executeSql(sql);

        if (list != null && !list.isEmpty()) {
            TimeValueVO result = list.get(0);
            Object value = result.getValue();

            // Check if the value is of type BigDecimal and then convert to Long
            if (value instanceof BigDecimal) {
                return ((BigDecimal) value).longValueExact();
            } else if (value instanceof Long) {
                return (Long) value;
            } else {
                // Handle the case where the value is not of an expected type
                log.error("Unexpected type for value: {}", value.getClass().getSimpleName());
                throw new IllegalArgumentException("Value is not of type BigDecimal or Long");
            }
        }
        return null; // Returns null if no records match the specific value
    }

    /**
     * 根据path聚合查询出一个值（avg,max,min等）
     *
     * @param path
     * @param startTime
     * @param endTime
     */
    public BigDecimal querySumValueIotDB(String path, long startTime, Long endTime, String aggregateMethod, BigDecimal th_low, BigDecimal th_high) {
        startTime = startTime * getTime();
        endTime = endTime * getTime();
        SessionDataSetWrapper sessionDataSet = null;
        try {

            String measurement = path.substring(path.lastIndexOf(".") + 1);
            String realPath = path.substring(0, path.lastIndexOf("."));
            String sql = null;
            if (th_high == null) {
                if (th_low == null) {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s", aggregateMethod, measurement, realPath, startTime, endTime);
                } else {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s >= %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_low);
                }
            } else {
                if (th_low == null) {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s <= %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_high);
                } else {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s >= %s and %s <= %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_low, measurement, th_high);
                }
            }
            sessionDataSet = iotdbSession.executeQueryStatement(sql);
            BigDecimal result = new BigDecimal(sessionDataSet.next().getFields().get(0).getStringValue());
            sessionDataSet.close();
            return result;
        } catch (Exception e) {
            log.error("查询IotDB失败", e);
            return BigDecimal.ZERO;
        } finally {
            if (sessionDataSet != null) {
                try {
                    sessionDataSet.close();
                } catch (Exception e) {
                    log.error("sessionDataSet资源为空，没有被正确关闭", e);
                    throw new BusinessException("sessionDataSet资源为空，没有被正确关闭");
                }
            }
        }
    }

    /**
     * 根据path聚合查询出一个值（avg,max,min等）
     *
     * @param path
     * @param startTime
     * @param endTime
     */
    public BigDecimal queryOpenValueIotDB(String path, long startTime, Long endTime, String aggregateMethod, BigDecimal th_low, BigDecimal th_high) {
        startTime = startTime * getTime();
        endTime = endTime * getTime();
        SessionDataSetWrapper sessionDataSet = null;
        try {

            String measurement = path.substring(path.lastIndexOf(".") + 1);
            String realPath = path.substring(0, path.lastIndexOf("."));
            String sql = null;
            if (th_high == null) {
                if (th_low == null) {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s", aggregateMethod, measurement, realPath, startTime, endTime);
                } else {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s > %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_low);
                }
            } else {
                if (th_low == null) {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s < %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_high);
                } else {
                    sql = String.format("select %s(%s) from %s where time >= %s and time <= %s and %s > %s and %s < %s", aggregateMethod, measurement, realPath, startTime, endTime, measurement, th_low, measurement, th_high);
                }
            }
            sessionDataSet = iotdbSession.executeQueryStatement(sql);
            BigDecimal result = new BigDecimal(sessionDataSet.next().getFields().get(0).getStringValue());
            sessionDataSet.close();
            return result;
        } catch (Exception e) {
            log.error("查询IotDB失败", e);
            return BigDecimal.ZERO;
        } finally {
            if (sessionDataSet != null) {
                try {
                    sessionDataSet.close();
                } catch (Exception e) {
                    log.error("sessionDataSet资源为空，没有被正确关闭", e);
                    throw new BusinessException("sessionDataSet资源为空，没有被正确关闭");
                }
            }
        }
    }


}

